package cn.jiaqiang.chatgpt.data.trigger.mq;

import cn.hutool.core.bean.BeanUtil;
import cn.jiaqiang.chatgpt.data.domain.order.service.IOrderService;
import cn.jiaqiang.chatgpt.data.types.common.RedisConstants;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ThreadPoolExecutor;


/**
 * User: jia-qiang
 * Date: 2024-04-06
 * Time: 13:38
 */
@Slf4j
@Component
@Deprecated
public class RedisStreamMq {

    @Resource
    private  StringRedisTemplate stringRedisTemplate;

    @Resource
    private  ThreadPoolExecutor threadPoolExecutor;

    @Resource
    private IOrderService orderService;

    /**
     * 提前读取好 判断购买资格的 lua 脚本
     */
    private static final DefaultRedisScript<Long> PAY_SUCCESS_SCRIPT;
    static {
        PAY_SUCCESS_SCRIPT = new DefaultRedisScript<>();
        // new ClassPathResource 就会在 resources 目录
        PAY_SUCCESS_SCRIPT.setLocation(new ClassPathResource("order_pay_success.lua"));
        // 设置返回值
        PAY_SUCCESS_SCRIPT.setResultType(Long.class);
    }

    @PostConstruct
    public void init() {
        // 加载该类的时候就会执行
        threadPoolExecutor.submit(new OrderHandler());
    }

    /**
     * 处理任务
     */
    private class OrderHandler implements Runnable {
        String queueName = RedisConstants.PAY_SUCCESS_QUEUE_NAME;
        String groupName = RedisConstants.PAY_SUCCESS_CONSUMER_GROUP;
        String consumer01 = RedisConstants.PAY_SUCCESS_CONSUMER_01;
        String orderIdKey = RedisConstants.PAY_SUCCESS_ORDER_ID_KEY;
        @Override
        public void run() {
            while (true) {
                // XREADGROUP GROUP g_openai c_consumer1 COUNT 1 BLOCK 2000 STREAMS k_pay_success
                List<MapRecord<String, Object, Object>> recordList = stringRedisTemplate.opsForStream().read(
                        Consumer.from(groupName, consumer01),
                        StreamReadOptions.empty().count(1).block(Duration.ofSeconds(5)),
                        StreamOffset.create(queueName, ReadOffset.lastConsumed())
                );
                // 2. 判断是否获取成功
                if (recordList == null || recordList.isEmpty()) {
                    // 2.1 失败, 说明消息队列中没有消息, 继续下一次循环
                    continue;
                }
                // 2.2 成功, 获取最新
                MapRecord<String, Object, Object> record = recordList.get(0); // count 1  --> get(0)
                // 获取键值对集合
                Map<Object, Object> value = record.getValue();
                try {
                    // 获取订单号
                    Object orderId = value.get(orderIdKey);
                    if (orderId == null) {
                        throw new RuntimeException("orderId error");
                    }
                    // 进行补货处理
                    log.info("支付完成，发货并记录，开始。订单：{}", orderId);
                    orderService.deliverGoods(orderId.toString());
                } catch (Exception e) {
                    // 重试机制尚未实现
                    log.error("支付完成, 发货出错: ", e);
                }
            }
        }
    }

    /**
     * 发布任务
     */
    public void post(String orderId) {
        // 执行 lua 脚本, 发布任务
       stringRedisTemplate.execute(
                // 指定 lua 脚本
                PAY_SUCCESS_SCRIPT,
                // 指定 KEYS 参数 -> 锁的key
                Collections.emptyList(),
                // 指定 ARGV 参数
               RedisConstants.PAY_SUCCESS_ORDER_ID_KEY, orderId
        );
    }

}
